package com.ebupt.migu.music.schedule.service.impl;

import com.ebupt.migu.music.schedule.entity.*;
import com.ebupt.migu.music.schedule.mapper.EffectStatisticMapper;
import com.ebupt.migu.music.schedule.service.EffectStatisticService;
import com.ebupt.migu.music.schedule.util.BeatchUtils;
import com.ebupt.migu.music.schedule.util.FtpUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.LocalDate;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author xuhongsong@cd.ebupt.com
 * @version 1.0
 * @className EffectStatisticServiceImpl
 * @date 2020/7/3 10:29
 */
@Slf4j
@Service
public class EffectStatisticServiceImpl implements EffectStatisticService {


    // 统计Mapper接口
    @Autowired
    private EffectStatisticMapper effectStatisticMapper;
    @Value("${msg.message_ftp_path}")
    private String paths;
    @Value("${msg.message_ftp_ip}")
    private String ip;
    // 用户名
    @Value("${msg.message_ftp_userName}")
    private String userName;
    // 密码
    @Value("${msg.message_ftp_userPwd}")
    private String userPwd;
    // 端口号
    @Value("${msg.message_ftp_ip_port}")
    private int port;
    // 读取文件的存放目录
    @Value("${msg.message_ftp_path}")
    private String telpath;

    @Resource
    private RedisTemplate redisTemplate;



    /**
     * 传入的i值去获取对应的文件名
     *
     * @param i
     * @return
     */
    public static String getFileName(int i, String yesterday) {
        String yesterdays = yesterday.substring(0, 10).replace("-", "");
        String paths = "message_";
        if (i < 10) {
            return paths.concat(yesterdays).concat("_00").concat(String.valueOf(i)).concat(".txt");
        } else if (i >= 10 && i < 100) {
            return paths.concat(yesterdays).concat("_0").concat(String.valueOf(i)).concat(".txt");
        } else {
            return paths.concat(yesterdays).concat("_").concat(String.valueOf(i)).concat(".txt");
        }
    }


    /**
     * 根据文件路径读取出来的文件转换成对应的List实体对象
     *
     * @return
     */
    private List<MessageNoticeInfo> MessageNoticeInfoList(String fileName, FtpUtils ftp) throws ParseException {
        List<String> names = ftp.readFile(fileName);
        // stream流转实体对象 返回2个参数  策列id 和读状态
        List<MessageNoticeInfo> messageNoticeInfoList = names.stream().
                map(e -> new MessageNoticeInfo(e.substring(0, e.indexOf("|")),
                        e.substring(e.lastIndexOf("|")))).collect(Collectors.toList());
        return messageNoticeInfoList;
    }

    /**
     * 去掉-的日期转化为日期
     *
     * @param date
     * @return
     */
    private String newString(String date) {
        return date.substring(0, 4).concat("-").concat(date.substring(4, 6)).concat("-").concat(date.substring(6));
    }

    /**
     * 根据策略Id 分组 并且对分组后的数据按照消息已读未读状态状态进行统计
     *
     * @param list
     * @return
     */
    private Map<String, Map<String, Long>> getGroupBystrategyId(List<MessageNoticeInfo> list) {
        //统计每个
        Map<String, Map<String, Long>> tradeNumMap = list.stream().
                collect(Collectors.groupingBy(MessageNoticeInfo::getTaskId,
                        Collectors.groupingBy(MessageNoticeInfo::getMsgStatus,
                                Collectors.counting())));
        return tradeNumMap;

    }

    /**
     * 转化为对应的List 做批量修改
     *
     * @return
     */
    public List<MsgUpdateBean> MsgUpdateList(Map<String, Map<String, Long>> map, String yesterday) {
        //  获取当前日期
        List<MsgUpdateBean> list = new ArrayList<>();
        //通过迭代器 循环
        Iterator<Map.Entry<String, Map<String, Long>>> iterator = map.entrySet().iterator();
        MsgUpdateBean msgUpdateBean;
        String read;
        String UnRead;
        String createDate;
        while (iterator.hasNext()) {
            Map.Entry<String, Map<String, Long>> entry = iterator.next();
            if (entry.getValue().containsKey("|0")) {
                UnRead = entry.getValue().get("|0").toString();
            } else {
                UnRead = "0";
            }
            if (entry.getValue().containsKey("|1")) {
                read = entry.getValue().get("|1").toString();
            } else {
                read = "0";
            }
            createDate = newString(entry.getKey().substring(entry.getKey().length() - 8));
            msgUpdateBean = new MsgUpdateBean(entry.getKey(), entry.getKey().substring(0, entry.getKey().length() - 8), read,
                    UnRead, createDate, yesterday);
            list.add(msgUpdateBean);
        }
        list = list.stream().filter(s -> s.getTaskId().substring(0, 2).equals("MN")).collect(Collectors.toList());
        //输出查找结果
        return list;
    }


    /**
     * 获取多个文件的List 集合
     *
     * @return
     */
    private List<MessageNoticeInfo> MessageNoticeInfoList(String yesterday) throws ParseException {
        // 首先判断当日的message_YYYYMMDD_001文件是否存在
        List<MessageNoticeInfo> list = new ArrayList<>();
        FtpUtils ftp = new FtpUtils();
        String path = telpath + yesterday.substring(0, 10).replace("-", "") + "/";
        log.info("跳转目录：{}", path);
        ftp.connectServer(ip, port, userName, userPwd, path);
        List<String> fileList = ftp.getFileList(path);
        Boolean IsHave = false;
        log.info("文件名:{}", getFileName(1, yesterday));
        if (isHaveFile(fileList, getFileName(1, yesterday))) {
            IsHave = true;
        }
        int i = 1;
        List<MessageNoticeInfo> messageNoticeInfoList;
        while (IsHave) {
            // 文件路径读取出来的文件转换成对应的List实体对象
            messageNoticeInfoList = MessageNoticeInfoList(getFileName(i, yesterday), ftp);
            if (messageNoticeInfoList != null && !messageNoticeInfoList.isEmpty()) {
                list.addAll(messageNoticeInfoList);
            }
            i++;
            IsHave = isHaveFile(fileList, getFileName(i, yesterday));
        }
        ftp.closeServer();
        return list;
    }

    /**
     * stream  判断是否存在该文件
     *
     * @param list
     * @param fileName
     * @return
     */
    public Boolean isHaveFile(List<String> list, String fileName) {
        return list.stream().filter(m -> m.equals(fileName)).findAny().isPresent();
    }

    /**
     * 执行的相关统计的任务具体业务
     *
     * @return
     */
    @Override
    public void RelatedStatistics(String yesterday) throws ParseException {
        // 获取多个文件的List 集合
        List<MessageNoticeInfo> list = MessageNoticeInfoList(yesterday);
        log.info("流水数据条数:{}", list.size());
        //  根据循环执行的结果集返回对应的 List<MsgUpdateBean>对象
        if (list != null && !list.isEmpty()) {
            // 1.根据策略Id 分组
            Map<String, Map<String, Long>> tradeNumMap = getGroupBystrategyId(list);
            // 2.转化为对应的List<MsgUpdateBean>对象
            List<MsgUpdateBean> MsgUpdateList = MsgUpdateList(tradeNumMap, yesterday);
            log.info("分组后统计结果数据条数:{}", MsgUpdateList.size());
            // 3.调用批量修改 分组进行批量结果集修改 mysql暂时为800
            List<List<MsgUpdateBean>> MsgList = new BeatchUtils<MsgUpdateBean>().beatchList(MsgUpdateList, 800);
            // 4.执行循环 批量修改
            int MsgListSize = MsgList.size();
            log.info("批量入库开始");
            Map<String, Object> map = new HashMap<>();
            for (int j = 0; j < MsgListSize; j++) {
                // 新增的是流水
                map.put("msgUpdateBeanList", MsgList.get(j));
                effectStatisticMapper.insertStatisticsReadStatus(map);
            }
            log.info("批量入库结束");
        }
    }

    /**
     * 全部修改入库
     */
    @Override
    public void update(String yesterday) {
        log.info("离线效果统计更新");
        // 获取流水统计返回结果
        List<StatisticsMsgBean> list = effectStatisticMapper.SelectMessage(yesterday);
        //  获取taskId 集合
        List<String> statisticsMsgList = effectStatisticMapper.selectStatisticsMsg();
        StatisticsMsg statisticsMsg;
        String taskId;
        String date1;
        int total;
        int singleRead;
        int singleUnread;
        int pv;
        int uv;
        int ldPv;
        int ldUv;
        String strategyId;
        List<StatisticsMsg> statisticsMsgAddList = new ArrayList<>();
        for (int i = 0, size = list.size(); i < size; i++) {
            //  taskId
            taskId = list.get(i).getTaskId();

            date1 = newString(taskId.substring(taskId.length() - 8));
            // 已读数
            singleRead = list.get(i).getSingleRead();
            // 未读数
            singleUnread = list.get(i).getSingleUnread();
            // 总数 针对添加  即第一次才会有
            total = singleRead + singleUnread;
            pv = list.get(i).getPv();
            uv = list.get(i).getUv();
            // 落地页pv统计
            ldPv = list.get(i).getLdPv();
            // 落地页uv统计
            ldUv = list.get(i).getLdUv();
            // 策略Id
            strategyId = taskId.substring(0, taskId.length() - 8);
            try {
                log.info("查询本地人群数开始！");
                // 本地人群
                Integer userIneger = effectStatisticMapper.selectUser(strategyId);
                log.info("本地人群数{}:", userIneger);
                int user = (null == userIneger ? 0 : userIneger);
                String s =yesterday.replaceAll("-","");
                String key_iop = strategyId + "_" + s + "_iop";
                String key_incr = strategyId + "_" + s + "_incr";
                String key_black = strategyId + "_" + s + "_black";
                log.info("redis数据拉取开始！");
                int inc = redisTemplate.hasKey(key_incr) == null ? 0
                        : null == redisTemplate.opsForValue().get(key_incr) ? 0
                        : Integer.valueOf(redisTemplate.opsForValue().get(key_incr).toString());
                log.info("inc:{}", inc);
                int intersectionUser = redisTemplate.hasKey(key_iop) == null ? 0
                        : (null == redisTemplate.opsForValue().get(key_iop) ? 0
                        : Integer.valueOf(redisTemplate.opsForValue().get(key_iop).toString()));
                log.info("intersectionUser:{}", intersectionUser);
                int blackUser = redisTemplate.hasKey(key_black) == null ? 0
                        : null == redisTemplate.opsForValue().get(key_black) ? 0
                        : Integer.valueOf(redisTemplate.opsForValue().get(key_black).toString());
                log.info("blackUser:{}", blackUser);
                int bdUser = user;
                log.info("bdUser:{}", bdUser);
                int totalFailure = (user - inc < 0 ? 0 : (user - inc));
                log.info("totalFailure:{}", totalFailure);
                statisticsMsg = new StatisticsMsg(strategyId, date1, total, singleRead, singleUnread, 0, 0,
                        yesterday, yesterday, ldPv, ldUv, taskId, 0, 0, intersectionUser, blackUser, bdUser, totalFailure, 0, 0, 0, 0);
                if (statisticsMsgList != null && !statisticsMsgList.isEmpty()) {
                    if (IsHave(statisticsMsgList, taskId)) {
                        //  修改
                        int code = effectStatisticMapper.update(statisticsMsg);
                        log.info("今日修改taskid:{},返回code:{}", taskId, code);
                    } else {
                        statisticsMsg.setCensus(1);
                        statisticsMsg.setPv(pv);
                        statisticsMsg.setUv(uv);
                        statisticsMsgAddList.add(statisticsMsg);
                    }
                } else {
                    statisticsMsg.setPv(pv);
                    statisticsMsg.setUv(uv);
                    statisticsMsg.setCensus(1);
                    statisticsMsgAddList.add(statisticsMsg);
                }
        }catch (Exception  exception){
         log.info("程序异常:{}", exception);
         System.out.println("程序异常:"+exception);
          }
        }

        //  批量新增
        if (statisticsMsgAddList != null && !statisticsMsgAddList.isEmpty()) {
            int code = effectStatisticMapper.insertBatch(statisticsMsgAddList);
            log.info("statisticsMsgAddList:{},今日新增taskid数:{}" + statisticsMsgAddList.size(), code);
        }
        int code = effectStatisticMapper.updateCallBack();
        int code1 = effectStatisticMapper.updatelist(yesterday);
        log.info("判断taskId任务是否10天结束执行条数code:{},修改流水的未读数和总数code:{}", code, code1);

        // 判断当天的数据是否有没有读取的  如果没有就新增一条
        List<String> taskIdList = effectStatisticMapper.taskIdList(yesterday);
        if (taskIdList != null && !taskIdList.isEmpty()) {
            int codes = effectStatisticMapper.insertBatchList(yesterday);
            Map<String, Object> map = new HashMap<>();
            map.put("taskIdList", taskIdList);
            map.put("yesterday", yesterday);
            int codes1 = effectStatisticMapper.updateBatchList(map);
            log.info("新增昨日未读taskId数code:{},修改主表taskId数code:{}", codes, codes1);
        }
        int code3 = effectStatisticMapper.updateCallBack();
        log.info("再次判断taskId任务是否10天结束执行条数code:{}", code3);
    }
 /**
     * 全部修改入库
     */
    @Override
    public void updateAuto(String yesterday) {
        log.info("自动运营效果统计更新");
        // 获取流水统计返回结果
        List<StatisticsMsgBean> list = effectStatisticMapper.SelectMessage(yesterday);
        //  获取taskId 集合
        List<String> autoMsgList = effectStatisticMapper.selectAutoMsg();
        AutoMsg autoMsg;
        String taskId;
        String date1;
        int total;
        int singleRead;
        int singleUnread;
        int pv;
        int uv;
        int ldPv;
        int ldUv;
        String tacId;
        List<AutoMsg> autoMsgAddList = new ArrayList<>();
        for (int i = 0, size = list.size(); i < size; i++) {
            //  taskId
            taskId = list.get(i).getTaskId();

            date1 = newString(taskId.substring(taskId.length() - 8));
            // 已读数
            singleRead = list.get(i).getSingleRead();
            // 未读数
            singleUnread = list.get(i).getSingleUnread();
            // 总数 针对添加  即第一次才会有
            total = singleRead + singleUnread;
            pv = list.get(i).getPv();
            uv = list.get(i).getUv();
            // 落地页pv统计
            ldPv = list.get(i).getLdPv();
            // 落地页uv统计
            ldUv = list.get(i).getLdUv();
            // 策略Id
            tacId = taskId.substring(0, taskId.length() - 8);
            try {
                autoMsg = new AutoMsg(tacId, date1, total, singleRead, singleUnread, 0, 0,
                        ldPv, ldUv, taskId );
                if (autoMsgList != null && !autoMsgList.isEmpty()) {
                    if (IsHave(autoMsgList, taskId)) {
                        //  修改
                        int code = effectStatisticMapper.updateAuto(autoMsg);
                        log.info("今日修改taskid:{},返回code:{}", taskId, code);
                    }else {
                        autoMsg.setPv(pv);
                        autoMsg.setUv(uv);
                        autoMsgAddList.add(autoMsg);
                    }
                }
                autoMsg.setPv(pv);
                autoMsg.setUv(uv);
                autoMsgAddList.add(autoMsg);
        }catch (Exception  exception){
         log.info("程序异常:{}", exception);
         System.out.println("程序异常:"+exception);
          }
        }
        //  批量新增
        if (autoMsgAddList != null && !autoMsgAddList.isEmpty()) {
            int code = effectStatisticMapper.insertBatchAuto(autoMsgAddList);
            log.info("autoMsgAddList:{},今日新增taskid数:{}" + autoMsgAddList.size(), code);
        }

    }



    /**
     * 计算转化率
     *
     * @return
     */
    @Override
    public void updateConversionRate() {
        effectStatisticMapper.updateConversionRate();
        effectStatisticMapper.updateAutoConversionRate();
    }

    /**
     * 判断今天是否执行了任务
     *
     * @return
     */
    @Override
    public Boolean IsHave() {
        return effectStatisticMapper.IsHave(LocalDate.now().minusDays(1).toString()) == 0 ? false : true;
    }

    @Override
    public boolean queryNewDate(String yesterday) {
        String newDate = effectStatisticMapper.queryNewDate();
        log.info("当前源数据库最新日期为：{}", newDate);
        return !newDate.equals(yesterday);
    }

    /**
     * 根据文件路径读取出来的文件转换成对应的List实体对象
     *
     * @return
     */
    private List<UnRelatedStatistics> UnRelatedStatisticsList(String fileName, FtpUtils ftp) {

        List<String> names = ftp.readFile(fileName);
        // stream流转实体对象 返回2个参数  策列id 和读状态
        List<UnRelatedStatistics> UnRelatedStatisticsList = names.stream().
                map(e -> new UnRelatedStatistics(e.substring(0, e.indexOf("|")),
                        e.substring(e.lastIndexOf("|") + 1))).collect(Collectors.toList());
        return UnRelatedStatisticsList;
    }

    /**
     * 执行的无关统计的任务具体业务
     *
     * @return
     */
    @Override
    public void UnRelatedStatistics(String yesterday) {
        //  获取昨天日期  并且去除-
        String yesterdays = yesterday.substring(0, 10).replace("-", "");
        String fileName = "msgPVUV_".concat(yesterdays).concat(".txt");
        FtpUtils ftp = new FtpUtils();
        String path = telpath + yesterday.substring(0, 10).replace("-", "") + "/";
        ftp.connectServer(ip, port, userName, userPwd, path);
        List<String> fileList = ftp.getFileList(path);
        log.info("判断msg里面文件个数:{}", fileList.size());
        if (isHaveFile(fileList, fileName)) {
            // 获取内容
            List<UnRelatedStatistics> unRelatedStatisticsList = UnRelatedStatisticsList(fileName, ftp);
            log.info("获取无关统计内容条数:{}", unRelatedStatisticsList.size());
            ftp.closeServer();
            // 写入库里
            if (unRelatedStatisticsList != null && !unRelatedStatisticsList.isEmpty()) {
                log.info("写入无关统计内容入库开始");
                Map<String, Object> map = new HashMap<>();
                map.put("unRelatedStatisticsList", unRelatedStatisticsList);
                map.put("date", yesterday);
                effectStatisticMapper.insertunRelatedStatisticsList(map);
                log.info("写入无关统计内容入库结束");
            }
        }
    }

    ;

    /**
     * 判断是否存在这个taskId
     *
     * @param statisticsMsgList
     * @param taskId
     * @return
     */
    private Boolean IsHave(List<String> statisticsMsgList, String taskId) {
        return statisticsMsgList.stream().filter(m -> m.equals(taskId)).findAny().isPresent();
    }


}
